/**
 * Copyright ©2003~2019年5月6日 szkingdom
 * @author raohj
 * @date 2019年5月6日
 * @file RTransferQueue.java
 * @comment : please input your comment to this file
 * @version : input version
 * @modify:
 * 		1、 2019年5月6日 由 raohj 新建文件
 */
package com.ccnu.concurrent.collection;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Date;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.TransferQueue;

/**
 * @author raohj
 * @date 2019年5月6日
 * @version 
 * @comment TransferQueue防止生产者过度生产，导致内存溢出。 通过transfer生产的消息，必须等待消费者调用take消费掉
 */
public class RTransferQueue {
	private static final Logger log = LoggerFactory.getLogger(RTransferQueue.class);
	
	/**
	 * @author raohj
	 * @date 2019年5月6日
	 * @version 
	 * @comment 生产者
	 */
	public static class RProduct implements Runnable{
		private static Random rd = new Random(new Date().getTime());
		private TransferQueue<String> msgQueue ;	// 消息队列
		/**
		 * 
		 */
		public RProduct(TransferQueue<String> msgQueue) {
			this.msgQueue = msgQueue;
		}
		@Override
		public void run() {
			for(int i=0; i<10; i++) {
				String msg = "Msg" + rd.nextInt();
				try {
					msgQueue.transfer(msg);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				log.info("消费完毕:" + msg);
			}
		}
	}
	
	/**
	 * @author raohj
	 * @date 2019年5月6日
	 * @version 
	 * @comment 消费者
	 */
	public static class RConsumer implements Runnable{
		private TransferQueue<String> msgQueue;	// 消息队列
		/**
		 * 
		 */
		public RConsumer(TransferQueue<String> msgQueue) {
			this.msgQueue = msgQueue;
		}
		@Override
		public void run() {
			while(msgQueue.size()>0) {
				try {
					log.info("消费者[" + Thread.currentThread().threadId() + "]:	" + msgQueue.take());
					Thread.sleep(100);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
	}
	
	public static void test() {
		TransferQueue<String> msgQueue = new LinkedTransferQueue<String>();	// 无界同步队列
		ExecutorService es = Executors.newFixedThreadPool(100);
		// 设置10个生产者
		for(int i=0; i<10; i++) {
			es.submit(new RProduct(msgQueue));
		}
		
		// 暂停10s等待产生消息
		try {
			Thread.sleep(3000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}

		// 设置5个消费者
		for(int i=0; i<5; i++) {
			es.submit(new RConsumer(msgQueue));
		}
		
		es.shutdown();
	}
}
